package user

import (
	"context"
	"fmt"
	"likeadmin/core"
	"strings"
	"time"

	"github.com/gin-gonic/gin"
	"github.com/unknwon/com"
	"gorm.io/gorm"
	"likeadmin/client/schemas/req"
	"likeadmin/core/response"
	"likeadmin/model/client"
	"likeadmin/util"
)

type ISListeningService interface {
	Oppo(c *gin.Context, req *req.ListeningOppoReq) (resp interface{}, e error)
	Oceanengine(c *gin.Context, req *req.ListeningOceanengineReq) (resp interface{}, e error)
	OppoTest(c *gin.Context, req *req.OppoTestReq) (resp interface{}, e error)
	Close()
}

// listeningService 结构体
type listeningService struct {
	db         *gorm.DB
	buffer     chan client.CallbackInfo
	batchSize  int
	ctx        context.Context
	cancelFunc context.CancelFunc
}

// NewListeningService 初始化函数
func NewListeningService(db *gorm.DB) ISListeningService {
	ctx, cancel := context.WithCancel(context.Background())
	service := &listeningService{
		db:         db,
		buffer:     make(chan client.CallbackInfo, 200),
		batchSize:  50,
		ctx:        ctx,
		cancelFunc: cancel,
	}

	// 启动后台处理协程
	go service.processBatch()

	return service
}

// Close 关闭监听服务
func (srv *listeningService) Close() {
	srv.cancelFunc()
	close(srv.buffer)
}

// processBatch 后台批量处理协程
func (srv *listeningService) processBatch() {
	ticker := time.NewTicker(30 * time.Second)
	defer ticker.Stop()

	batch := make([]client.CallbackInfo, 0, srv.batchSize)

	for {
		select {
		case <-srv.ctx.Done():
			// 处理剩余批次
			if len(batch) > 0 {
				srv.batchInsert(batch)
			}
			return
		case callbackInfo := <-srv.buffer:
			batch = append(batch, callbackInfo)
			if len(batch) >= srv.batchSize {
				srv.batchInsert(batch)
				batch = batch[:0]
			}
		case <-ticker.C:
			if len(batch) > 0 {
				srv.batchInsert(batch)
				batch = batch[:0]
			}
		}
	}
}

// batchInsert 批量插入方法
func (srv *listeningService) batchInsert(batch []client.CallbackInfo) {
	if len(batch) == 0 {
		return
	}

	err := srv.db.Model(client.CallbackInfo{}).CreateInBatches(batch, len(batch)).Error
	if err != nil {
		core.Logger.Warnf("批量插入失败: %v", err)
	} else {
		core.Logger.Infof("批量插入成功，共 %d 条数据", len(batch))
	}
}

// saveCallbackInfo 保存回调信息的通用方法
func (srv *listeningService) saveCallbackInfo(callbackInfo client.CallbackInfo, redisKey string) error {
	// 检查Redis中是否存在该键值（10分钟内是否有记录）
	exists := util.RedisUtil.Exists(redisKey)
	if exists > 0 {
		return nil // 10分钟内有重复记录，直接返回
	}

	// 将数据发送到缓冲通道（非阻塞）
	select {
	case srv.buffer <- callbackInfo:
		core.Logger.Debugf("数据已发送到缓冲通道: %s", redisKey)
	default:
		// 通道满时的处理
		core.Logger.Debugf("缓冲通道已满，直接入库: %s", redisKey)
		err := srv.db.Model(client.CallbackInfo{}).Create(&callbackInfo).Error
		if err != nil {
			return fmt.Errorf("单条插入失败: %w", err)
		}
	}

	// 在Redis中设置键值，有效期10分钟
	util.RedisUtil.Set(redisKey, "1", 600)
	return nil
}

// getUAPrefix 获取UA前缀
func getUAPrefix(ua string, maxLength int) string {
	ua = strings.ToLower(ua)
	if len(ua) > maxLength {
		return ua[:maxLength]
	}
	return ua
}

// Oppo 方法
func (srv *listeningService) Oppo(c *gin.Context, req *req.ListeningOppoReq) (resp interface{}, e error) {
	// 将结构体转换为JSON字符串
	linkJson, err := util.ToolsUtil.ObjToJson(req)
	if err != nil {
		e = response.CheckErr(err, "转换JSON失败")
		return
	}

	// 获取UA前缀
	uaPrefix := getUAPrefix(req.UserAgent, 32)

	// 构造Redis键值
	redisKey := fmt.Sprintf("backstage:client:oppo_callback:%s:%s", req.IP, uaPrefix)

	// 创建回调信息对象
	callbackInfo := client.CallbackInfo{
		Channel:  4, // OPPO渠道
		SourceId: req.SpreadId,
		Ip:       req.IP,
		Ua:       req.UserAgent,
		Link:     linkJson,
	}

	// 保存回调信息
	if err := srv.saveCallbackInfo(callbackInfo, redisKey); err != nil {
		e = response.CheckErr(err, "保存回调信息失败")
		return
	}

	return "", nil
}

// Oceanengine 头条监测链接处理
func (srv *listeningService) Oceanengine(c *gin.Context, req *req.ListeningOceanengineReq) (resp interface{}, e error) {
	// 将结构体转换为JSON字符串
	linkJson, err := util.ToolsUtil.ObjToJson(req)
	if err != nil {
		e = response.CheckErr(err, "转换JSON失败")
		return
	}

	// 获取UA前缀
	uaPrefix := getUAPrefix(req.Ua, 32)

	// 构造Redis键值
	redisKey := fmt.Sprintf("backstage:client:oceanengine_callback:%s:%s", req.IP, uaPrefix)

	// 创建回调信息对象，包含头条特有的项目信息
	callbackInfo := client.CallbackInfo{
		Channel:       1, // 头条渠道
		SourceId:      req.SpreadId,
		Ip:            req.IP,
		Ua:            req.Ua,
		Link:          linkJson,
		ProjectId:     req.ProjectId,     // 项目ID
		ProjectName:   req.ProjectName,   // 项目名称
		PromotionId:   req.PromotionId,   // 推广计划ID
		PromotionName: req.PromotionName, // 推广计划名称
	}

	// 保存回调信息
	if err := srv.saveCallbackInfo(callbackInfo, redisKey); err != nil {
		e = response.CheckErr(err, "保存回调信息失败")
		return
	}

	return "", nil
}

// OppoTest 测试方法
func (srv *listeningService) OppoTest(c *gin.Context, req *req.OppoTestReq) (resp interface{}, e error) {
	// 查询回调信息
	var callbackInfo client.CallbackInfo
	err := srv.db.Where("id = ? AND callback_status = ?", req.Id, 0).Limit(1).First(&callbackInfo).Error
	if e = response.CheckErrDBNotRecord(err, "callbackInfo First err"); e != nil {
		return
	}

	// 解析链接数据为对象
	var reportData map[string]interface{}
	err = util.ToolsUtil.JsonToObj(callbackInfo.Link, &reportData)
	if err != nil {
		e = response.CheckErr(err, "回调数据解析失败")
		return
	}

	// 检查解析结果
	if reportData == nil {
		e = response.AssertArgumentError.Make("回调数据解析结果为空")
		return
	}

	// 准备上报数据
	reportData["id"] = com.ToStr(callbackInfo.ID)
	reportData["dataType"] = 16
	reportData["package_name"] = "com.miniw.liaolie01.xinwang"

	// 回传上报
	result := util.OppoCallbackUtil{}.UploadOppo(reportData)

	core.Logger.Infof("OPPO回调上报结果: %s", result)

	// 检查上报结果
	if strings.Contains(result, "::error") {
		e = response.CheckErr(fmt.Errorf("upload failed"), "更新回调状态失败:id=%d", reportData["id"])
		return
	}

	// 更新回调状态
	callbackInfo.CallbackStatus = 1 // 标记为已回调
	callbackInfo.UpdateTime = time.Now().Unix()

	err = srv.db.Save(&callbackInfo).Error
	if err != nil {
		e = response.CheckErr(err, "更新回调状态失败")
		return
	}

	return result, nil
}
